/**
 * Copyright [2019] [LiBo/Alex of copyright liboware@gmail.com ]
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hyts.standard.webflux.sample.flux;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Stream;

/**
 * @project-name:standard-webflux
 * @package-name:com.hyts.standard.webflux.sample.load.flux
 * @author:LiBo/Alex
 * @create-date:2020-05-15 14:46
 * @copyright:libo-alex4java
 * @email:liboware@gmail.com
 * @description:
 */
public class FluxSample {


    /**
     * 加载数据信息
     * @param objects
     * @param consumer
     */
    public static<T> void install(T objects,Consumer<T> consumer){
        Flux.just(objects).subscribe(consumer);
    }

    /**
     * 加载数据信息
     * @param objects
     * @param consumer
     */
    public static<T> void install(T[] objects,Consumer<T> consumer){
        Flux.fromArray(objects).subscribe(consumer);
    }

    /**
     * 加载数据信息
     * @param objects
     * @param <T>
     */
    public static<T> void install(Consumer<T> consumer,T... objects){
        Flux.just(objects).subscribe(consumer);
    }

    /**
     * 加载数据信息
     * @param collection
     * @param <T>
     */
    public static<T> void install(Collection<T> collection,Consumer<T> consumer){
        Flux.fromIterable(collection).subscribe(consumer);
    }

    /**
     * 加载数据信息
     * @param stream
     * @param <T>
     */
    public static<T> void install(Stream<T> stream,Consumer<T> consumer){
        Flux.fromStream(stream).subscribe(consumer);
    }

    /**
     * 加载数据信息
     * @param stream
     * @param <T>
     */
    public static<T> void install(Supplier<Stream<? extends T>> stream,Consumer<T> consumer){
        Flux.fromStream(stream).subscribe(consumer);
    }




    public static void main(String[] args){
        //FluxSample.install(System.out::println,12,1231,1231);
        try {
            testReactor2();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    public static void testReactor(){
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6);
        Mono<Integer> mono = Mono.just(1);

        Integer[] arr = {1,2,3,4,5,6};
        Flux<Integer> flux1 = Flux.fromArray(arr);

        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6);
        Flux<Integer> flux2 = Flux.fromIterable(list);

        Flux<Integer> flux3 = Flux.from(flux);

        Flux<Integer> flux4 = Flux.fromStream(Stream.of(1, 2, 3, 4, 5, 6));

//        flux.subscribe();
//
//        flux1.subscribe(System.out::println);
//
//        flux2.subscribe(System.out::println,System.err::println);
//
//        flux3.subscribe(System.out::println,System.err::println,() -> System.out.println("complete"));

//        flux4.subscribe(System.out::println,System.err::println,
//                () -> System.out.println("complete"),
//                subscription -> subscription.request(2));

        flux4.subscribe(new DemoSubscriber());
    }

    static class DemoSubscriber extends BaseSubscriber<Integer> {
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            System.out.println("Subscribe");
            subscription.request(3);
        }

        @Override
        protected void hookOnNext(Integer value) {
            if(value == 5){
                //背压，通知数据源，不要发送数据了
                cancel();
            }
            System.out.println(value);
           // request(6);
        }
    }

//    线程切换总结
//    publishOn：它将上游信号传给下游，同时改变后续的操作符的执行所在线程，直到下一个publishOn出现在这个链上
//    subscribeOn：作用于向上的订阅链，无论处于操作链的什么位置，它都会影响到源头的线程执行环境，但不会影响到后续的publishOn

    public static void testReactor2() throws InterruptedException {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6);
        flux.map(i -> {
            System.out.println(Thread.currentThread().getName()+"-map1");
            return i * 3;
        }).publishOn(Schedulers.elastic()).map(
                i -> {
                    System.out.println(Thread.currentThread().getName()+"-map2");
                    return i / 3;
                }
        ).subscribeOn(Schedulers.parallel())
                .subscribe(i -> System.out.println(Thread.currentThread().getName()+"-" + i));

        Thread.sleep(10000);
    }

}
